#ifndef __DISKCACHE_HH__
#define __DISKCACHE_HH__
#include "ptypes.h"
#include "log.h"
#include "drecodec.h"
#include "chunk.h"
#include "diskcache.h"
#include <iostream>
#include <map>
#include <assert.h>
#include "cacheindex.h"
#include "chunk.h"
#include "chunkstats.h"

#include <fcntl.h> // for open
#include <string.h> // for memcpy

// file structure:
// <header>

static const unsigned long DCH_SIG = 0xBEDADBEA;

typedef struct {
  unsigned long signature;
  unsigned int bits;
  unsigned int guardRecords;
  unsigned int quarantinRecords;

  unsigned int inboundLastRecord;
  unsigned int inboundFirstRecord;
  bool inboundFirstSave;

  unsigned int outboundFirstRecord;
  unsigned int outboundLastRecord;
  bool outboundFirstSave;

} DiskCacheHeader;

typedef std::map<int, int> GenMap;
typedef std::pair<int, int> ChunkGen;
typedef std::pair<int, unsigned long> ChunkUsage;
typedef std::map<ChunkGen, ChunkUsage> ChunkStats;

class DiskCache {
  friend class Packeteer;
  long m_handle;
  int m_context;
  bool m_firstTime;
  DiskCacheHeader m_header;
  int m_half, m_oneWayRecords, m_recordSize;

  float m_redundancyThreshold;
  float m_purgingPoolThreshold;

  bool m_readOnlyCache;

  int m_penalty;

  bool m_rawdiskMode, m_memoryMode;
  int m_rawBlocks, m_blockSize;

  class Packeteer *m_parent;

  // stats
  unsigned long m_purges;
  unsigned long m_gets;
  unsigned long m_saves;

  bool m_collectChunkStats;
  ChunkStats m_ChunkStats;
  GenMap m_ChunkGenerations;

  ChunkStatistics *m_runtimeStats;
 public:
  friend class Chunk;
  CacheIndex m_index;
  DiskCache();
 ~DiskCache();

 bool Initialize(class Packeteer *parent, int context, const char *raw_name);
 
 // returns 1 - record exists
 // returns -1 - record exists in quarantine
  inline int RecordStatus(unsigned int recordNumber) {
#if 0
    int delta;

    if(recordNumber >= (unsigned)m_oneWayRecords)
	delta =  recordNumber - m_header.outboundFirstRecord;
    else
	delta = recordNumber - m_header.inboundFirstRecord;
    delta = delta % m_oneWayRecords;
    if(delta < 0)
	delta += m_oneWayRecords;

    if(delta < m_header.quarantinRecords + m_header.guardRecords)
	return -1;
#endif
    return 1;
  }

  // find a record for fingerprint
  inline bool FindFingerprint(Fingerprint f, int *recordnumber) {
    bool ret = m_index.FindFingerprint(f, recordnumber);
    if(!ret)
	return false;

    if(RecordStatus(*recordnumber) < 0)
	return false; 

    return true;
  }

  inline void PurgePacket(int packetID) {
    Chunk *chunk = GetChunk(packetID);
    //Log("Delete packet %d ", packetID);
    // walk the fingerprints, and remove them from the main index
    chunk->Rewind();
    Fingerprint fp;
    short pos;
Log ("Purge: index %d\n", packetID);
    while(chunk->GetNextFingerprint(&fp, &pos)) {
Log ("    fingerprint %16llx\n", fp);
	m_index.DeleteFingerprint(&fp);
    }

    delete chunk;
    RecordChunkPurge(packetID);
    m_purges++;
  }

  // retrieve a chunk by its record number 
  inline class Chunk *GetChunk(unsigned int recordNumber) {
    assert(m_handle != -1);
    // TODO - see if it is in the range..
    Chunk *chunk = new Chunk(m_parent);
    long offset;
    if (m_rawdiskMode) {
      offset = 2*sizeof(m_header);
      offset += 2 * m_half * m_context;
      offset += chunk->GetSizeOnDisk()*recordNumber;
    }
    else
      offset = sizeof(m_header) + chunk->GetSizeOnDisk()*recordNumber;
    //printf("GetChunk(%d): using %ld for record %d\n", m_context, offset, recordNumber);
    cacheLseek(offset, SEEK_SET);
    chunk->Deserialize(m_handle);
    m_gets++;
    return chunk;
  }


  // save a chunk, return a purged chunk /TODO/
  inline bool SaveChunk(Chunk *inchunk, unsigned int *recordNumber, ReDirection way) {
    int newRecordNumber, firstRecord;
    int delta;
    long offset;

    if(way == ReOutbound) {
      // m_header.outboundFirstSave now indicates first full fill of the m_oneWayRecords-sized pool
      if (m_header.outboundFirstSave) {
        // m_header.outboundLastRecord is initialized to -1 elsewhere
        newRecordNumber = m_header.outboundLastRecord + 1 + m_oneWayRecords;
        m_header.outboundLastRecord++;
        if (m_header.outboundLastRecord == (m_oneWayRecords-1)) {
          // we reached the end, different mode now
          m_header.outboundFirstSave = false;
          if (m_purgingPoolThreshold > 0.0) {
            m_header.outboundLastRecord = m_oneWayRecords - int((float)m_oneWayRecords *m_purgingPoolThreshold);
          }
          else // old-style
            m_header.outboundLastRecord = -1;
        }
      }
      else {
        // LRU buffer
        m_header.outboundLastRecord++;
        if (m_purgingPoolThreshold > 0.0) { // new style
          if (m_header.outboundLastRecord == (m_oneWayRecords-1))
            m_header.outboundLastRecord = m_oneWayRecords - int((float)m_oneWayRecords *m_purgingPoolThreshold);
          int distance_from_bottom = m_oneWayRecords - m_header.outboundLastRecord - 1;
          int victim = m_runtimeStats->GetBottom(distance_from_bottom);
          assert(victim != -1);
          //std::cout << "Purging " << victim+m_oneWayRecords << " at distance " << distance_from_bottom << std::endl;
	  PurgePacket(victim + m_oneWayRecords);
          newRecordNumber = victim + m_oneWayRecords;
        }
        else { // old-style
          int victim = m_header.outboundLastRecord;
          PurgePacket(victim + m_oneWayRecords);
          //std::cout << "Purging " << victim+m_oneWayRecords << " at position " << victim << std::endl;
          newRecordNumber = victim + m_oneWayRecords;
          if (m_header.outboundLastRecord == (m_oneWayRecords-1))
            m_header.outboundLastRecord = -1;
        }
      }

      if (m_rawdiskMode) {
        offset = 2*sizeof(m_header) + 2 * m_half * m_context;
        offset += newRecordNumber * inchunk->GetSizeOnDisk();
      }
      else
	offset = sizeof(m_header) + newRecordNumber * inchunk->GetSizeOnDisk();

    } else {
      assert(way == ReInbound);
      if (m_header.inboundFirstSave) {
        // m_header.inboundLastRecord is initialized to -1 elsewhere
        newRecordNumber = m_header.inboundLastRecord + 1;
        m_header.inboundLastRecord++;
        if (m_header.inboundLastRecord == (m_oneWayRecords-1)) {
          // we reached the end, different mode now
          m_header.inboundFirstSave = false;
          if (m_purgingPoolThreshold > 0.0) {
            m_header.inboundLastRecord = m_oneWayRecords - int((float)m_oneWayRecords *m_purgingPoolThreshold);
          }
          else // old-style
            m_header.inboundLastRecord = -1;
        }
      }
      else {
        // LRU buffer
        m_header.inboundLastRecord++;
        if (m_purgingPoolThreshold > 0.0) { // new style
          if (m_header.inboundLastRecord == (m_oneWayRecords-1))
            m_header.inboundLastRecord = m_oneWayRecords - int((float)m_oneWayRecords *m_purgingPoolThreshold);
          int distance_from_bottom = m_oneWayRecords - m_header.inboundLastRecord - 1;
          // TODO: need a separate runtime stat structure for inbound
          int victim = m_runtimeStats->GetBottom(distance_from_bottom);
          assert(victim != -1);
          //std::cout << "Purging " << victim << " at distance " << distance_from_bottom << std::endl;
	  PurgePacket(victim);
          newRecordNumber = victim;
        }
        else { // old-style
          int victim = m_header.inboundLastRecord;
          PurgePacket(victim);
          //std::cout << "Purging " << victim << " at position " << victim << std::endl;
          newRecordNumber = victim;
          if (m_header.inboundLastRecord == (m_oneWayRecords-1))
            m_header.inboundLastRecord = -1;
        }
      }

      if (m_rawdiskMode) {
        offset = 2*sizeof(m_header) + 2 * m_half * m_context;
        offset += newRecordNumber * inchunk->GetSizeOnDisk();
      }
      else
	offset = sizeof(m_header) + newRecordNumber * inchunk->GetSizeOnDisk();
    }

    *recordNumber = newRecordNumber;
    inchunk->m_recordNumber = newRecordNumber;
    cacheLseek(offset, SEEK_SET);

    inchunk->Serialize(m_handle, &m_index, newRecordNumber);

    m_saves++;
  
    return true;
  }

  unsigned long GetStatsPurges() { return m_purges; }
  unsigned long GetStatsGets() { return m_gets; }
  unsigned long GetStatsSaves() { return m_saves; }

  inline void RecordChunkPurge(int packetNumber) {
    if (m_collectChunkStats) {
      GenMap::iterator it = m_ChunkGenerations.find(packetNumber);
      if (it == m_ChunkGenerations.end())
        m_ChunkGenerations[packetNumber] = 1;
      else
        it->second++;
    }
    if (m_runtimeStats) {
      ChunkStatistics::Usage u = { 0, 0L };
      // TODO: need a separate runtime stat structure for inbound
      m_runtimeStats->RecordStats(packetNumber >= m_oneWayRecords ? packetNumber-m_oneWayRecords:packetNumber, u);
      //assert(m_runtimeStats->ValidateStats());
    }
  }

  inline void RecordChunkStats(int packetNumber, int spanLength) {
    if (m_collectChunkStats) {
      // first find out what the current generation of the chunk is
      GenMap::iterator itg = m_ChunkGenerations.find(packetNumber);
      int gen = (itg == m_ChunkGenerations.end() ? 0 : itg->second);
      ChunkStats::iterator it = m_ChunkStats.find(ChunkGen(packetNumber, gen));
      if (it == m_ChunkStats.end())
        m_ChunkStats[std::pair<int,int>(packetNumber,gen)] = ChunkUsage(1, (unsigned long)spanLength);
      else {
        it->second.first++;
        it->second.second += (unsigned long)spanLength;
      }
    }
    if (m_runtimeStats) {
      ChunkStatistics::Usage u = { 1, spanLength };
      // TODO: need a separate runtime stat structure for inbound
      m_runtimeStats->AccumStats(packetNumber >= m_oneWayRecords ? packetNumber-m_oneWayRecords:packetNumber, u);
      //assert(m_runtimeStats->ValidateStats());
    }
  }

  inline void PrintChunkStats(bool emitFPMap, bool emitChunkFP) {
    if (m_collectChunkStats) {
      std::cout << "Disk cache chunk stats:" << std::endl;
      std::cout << "Chunk map size: " << m_ChunkStats.size() << std::endl;
      for (ChunkStats::iterator it = m_ChunkStats.begin();
	   it != m_ChunkStats.end(); it++) {
        ChunkGen chg = it->first;
        std::cout << "Chunk: " << chg.first << ", generation " << chg.second <<
          ", hit count: " << it->second.first <<
          ", byte count: " << it->second.second <<
          ", ratio: " << (float)it->second.second/(float)it->second.first <<
          std::endl;
        if (emitChunkFP) {
          Chunk *chunk = GetChunk(chg.first);
          assert(chunk);
          Fingerprint fp;
          short pos;
          while(chunk->GetNextFingerprint(&fp, &pos)) {
            std::cout << "Fingerprint: " << fp << ", " << "chunk number: " << chg.first << std::endl;
          }
          delete chunk;
        }
      }
      if (emitFPMap)
        m_index.PrintFingerprintMap();
    }
    if (m_runtimeStats) {
      // TODO: need a separate runtime stat structure for inbound
      m_runtimeStats->PrintStats(true); // from top to bottom
    }
  }

  inline int cacheRead(void *data, int size ) {
   if (m_memoryMode) {
     unsigned char *payload = (unsigned char *)(m_handle + sizeof(long));
     long offset = *((long *)m_handle);
     memcpy(data, payload + offset, size);
     return size;
   } else
     return read(m_handle, data, size);
 }

  inline int cachePread(void *data, int size, off_t offset ) {
   if (m_memoryMode) {
     unsigned char *payload = (unsigned char *)(m_handle + sizeof(long));
     memcpy(data, payload + offset, size);
     return size;
   } else
     return pread(m_handle, data, size, offset);
  }
 
  inline int cacheWrite(void *data, int size ) {
   if (m_memoryMode) {
     unsigned char *payload = (unsigned char *)(m_handle + sizeof(long));
     long offset = *((long *)m_handle);
     memcpy(payload + offset, data, size);
     return size;
   } else
     return write(m_handle, data, size);
  }
 
  inline int cachePwrite(void *data, int size, off_t offset ) {
   if (m_memoryMode) {
     unsigned char *payload = (unsigned char *)(m_handle + sizeof(long));
     memcpy(payload + offset, data, size);
     return size;
   } else
     return pwrite(m_handle, data, size, offset);
  }
 
  inline long cacheOpen(char *name, int mode, int perm) {
     return open(name, mode, perm);
  }
 
  inline int cacheLseek(long pos, int how ) {
   if (m_memoryMode) {
     long *ppos = (long *)m_handle;
     *ppos = pos;
     return pos;
   } else
     return lseek(m_handle, pos, how);
  }

  inline void cacheClose() {
   if (m_memoryMode) {
     delete (unsigned char *)m_handle;
   } else
     close(m_handle);
  }

};


#endif
